package com.mjf.day3

import java.util

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests

/**
 * Flink写入到ES
 */
object MyEsSink {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\flink-stu\\src\\main\\input\\sensor.txt")

    val dataStream: DataStream[SensorReading] = inputStream.map(
      data => {
        val dataArr: Array[String] = data.split(",")
        SensorReading(dataArr(0), dataArr(1).toLong, dataArr(2).toDouble)
      }
    )

    // 定义一个httpHosts
    val hosts: util.ArrayList[HttpHost] = new util.ArrayList[HttpHost]()
    hosts.add(new HttpHost("hadoop103", 9200))
    // 定义一个ElasticsearchSinkFunction
    val esSinkFunction: ElasticsearchSinkFunction[SensorReading] = new ElasticsearchSinkFunction[SensorReading]() {
      override def process(element: SensorReading, ctx: RuntimeContext, indexer: RequestIndexer): Unit = {
        // 包装写入ES的数据
        val dataSource: util.HashMap[String, String] = new util.HashMap[String, String]()
        dataSource.put("sensor_id", element.id)
        dataSource.put("sensor_temp", element.temperature.toString)
        dataSource.put("sensor_ts", element.timestamp.toString)
        // 创建一个index request
        val indexRequest: IndexRequest = Requests.indexRequest()
          .index("sensor_temp")
          .`type`("readingData")
          .source(dataSource)
        // 用indexer发送http请求
        indexer.add(indexRequest)
      }
    }

    // 数据保存到ES
    dataStream.addSink(
      new ElasticsearchSink.Builder[SensorReading](hosts, esSinkFunction).build()
    )

    env.execute("MyEsSink")

  }
}
